package com.xinqing.bigdata.flink.datastream.source;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author:CHQ
 * @Date:2021/4/15 10:41
 * @Description
 */
public class Sources {
    public static void main(String args[]) throws Exception {
        //env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //source from POJO
      /*  DataStream<Person> source = env.fromElements(
                new Person("chq", 23)
                , new Person("jay", 44));*/

        //source from collection
        List<Person> list = new ArrayList<>();
        list.add(new Person("chq", 22));
        list.add(new Person("jay", 44));

        DataStream<Person> source = env.fromCollection(list);

        //transform
        DataStream<Person> filter = source.filter((FilterFunction<Person>) value -> value.age > 24);

        //sink
        filter.print();

        //excute
        env.execute("example");
    }

    private static class Person {
        String name;
        int age;

        public Person(String name, int age) {
            this.name = name;
            this.age = age;
        }

        @Override
        public String toString() {
            return "Person{" +
                    "name='" + name + '\'' +
                    ", age=" + age +
                    '}';
        }
    }
}


